package net.bwie.dws;

import javafx.scene.control.Tab;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.TableEnvironment;
import org.apache.flink.table.api.TableResult;

//dws_refund_time（退款时间表）

public class Dws_refund_time {
    public static void main(String[] args) {

        // 1.创建环境
        TableEnvironment tableEnv = getTableEnv();

        // 2.创建输入表
        createInputTable(tableEnv);


        // 3.创建输出表
        createoutputTable(tableEnv);

        // 4.处理数据
        Table resultTable = handle(tableEnv);

//        tableEnv.createTemporaryView("resultTable", resultTable);
//        tableEnv.executeSql("SELECT * FROM resultTable ").print();

        // 5.保存数据
        saveToSink(tableEnv, resultTable);
    }

    private static TableEnvironment getTableEnv() {
        EnvironmentSettings build = EnvironmentSettings.newInstance()
                .inStreamingMode()
                .build();
        TableEnvironment tableEnvironment = TableEnvironment.create(build);

        Configuration configuration = tableEnvironment.getConfig().getConfiguration();
        configuration.setString("table.local-time-zone", "Asia/Shanghai");
        configuration.setString("table.exec.resource.default-parallelism", "1");
        configuration.setString("table.exec.state.ttl", "5 s");
        configuration.setString("execution.checkpointing.interval", "5 s");

        return tableEnvironment;
    }

    private static void createInputTable(TableEnvironment tableEnv) {

        tableEnv.executeSql("CREATE TABLE dwd_retreat_behavior (\n" +
                //订单id
                "order_id STRING ,\n" +
                //  商品ID
                "product_id STRING ,\n" +
                // 订单支付时间
                "pay_time STRING ,\n" +

                // 退款单号
                "refund_id STRING ,\n" +
                // 退款申请时间
                "apply_time TIMESTAMP(3) ,\n" +
                //  退款场景:“退货退款”“已收货仅退款”“未收货退款”“未发货仅退款”
                "refund_scene STRING ,\n" +
                // 实际退款金额
                "refund_amount DOUBLE ,\n" +
                // 退款状态
                "refund_status STRING,\n"+

                //消费者行为id
                "behavior_id STRING ,\n" +
                //消费者id
                "user_id STRING ,\n" +
                //行为类型
                "behavior_type STRING, \n" +

                " time_ltz AS TO_TIMESTAMP_LTZ(UNIX_TIMESTAMP(pay_time),0),\n" +
                "WATERMARK FOR time_ltz AS time_ltz - INTERVAL '5' DAY\n" +
                ")WITH(\n" +
                "'connector' = 'kafka',\n" +
                "'topic' = 'dwd_retreat_behavior',\n" +
                "'properties.bootstrap.servers' = 'node101:9092',\n" +
                "'properties.group.id' = 'dwd_retreat_behavior-id06',\n" +
                "'scan.startup.mode' = 'earliest-offset',\n" +
                "'format' = 'json'\n" +
                ")"
        );


//        TableResult tableResult4 = tableEnv.executeSql("SELECT * FROM dwd_retreat_behavior " +
//                " WHERE behavior_id IS NOT NULL and refund_id IS NOT NULL ");
//        tableResult4.print();

    }

    private static void createoutputTable(TableEnvironment tableEnv) {
        //dws_refund_after_behavior
        tableEnv.executeSql("CREATE TABLE dws_refund_after_behavior (\n" +
                // 窗口开始时间
                "window_start TIMESTAMP(3),\n" +
                // 窗口结束时间
                "window_end TIMESTAMP(3),\n" +

                //退款时间类型 支付30分钟内、30分钟-24小时、24小时以上
                "refund_time_type STRING ,\n" +

                //订单数
                "order_num INT\n"+

                ")WITH(\n" +
                        " 'connector' = 'doris',\n" +
                        "'fenodes' = 'node102:8030',\n" +
                        "'table.identifier' = 'transactions_dws.dws_refund_time',\n" +
                        "'username' = 'root',\n" +
                        "'password' = '123456',\n" +
                        "'sink.label-prefix' = 'dws_refund_time_id06')\n"
        );
    }

    private static Table handle(TableEnvironment tableEnv) {

        Table tableResult = tableEnv.sqlQuery(
                "SELECT\n" +
                        "window_start,\n" +
                        "window_end,\n" +
                        "refund_time_type,\n" +
                        "CAST(COUNT(order_id) as INT) as order_num\n" +
                        "FROM (\n" +
                            "SELECT \n" +
                            "window_start,\n" +
                            "window_end,\n" +
                            "CASE \n" +
                            "   WHEN diff_minutes<=30 THEN '支付30分钟内'\n" +
                            "   WHEN diff_minutes>30 and diff_hours<=24 THEN '30分钟-24小时'\n" +
                            "   ELSE '24小时以上'\n" +
                            "END AS refund_time_type,\n" +
                            "order_id\n" +
                            "FROM (\n" +
                                "SELECT \n" +
                                "window_start\n" +
                                ",window_end\n" +
                                ",order_id\n" +
                                ",time_ltz\n" +
                                ",apply_time\n" +
                                ",TIMESTAMPDIFF(MINUTE, time_ltz, CAST(apply_time AS TIMESTAMP_LTZ(3))) AS diff_minutes\n" +
                                ",TIMESTAMPDIFF(HOUR, time_ltz, CAST(apply_time AS TIMESTAMP_LTZ(3))) AS diff_hours\n" +
                                    "FROM TABLE(\n" +
                                    // 添加 1天 的 滚动窗口
                                    "  TUMBLE(TABLE dwd_retreat_behavior, DESCRIPTOR(time_ltz), INTERVAL '1' day)\n" +
                                    ")\n" +
                                "WHERE time_ltz IS NOT NULL AND apply_time IS NOT NULL and " +
                        "refund_id IS NOT NULL and behavior_id  IS NOT NULL \n" +
                                ")\n" +
                            ")\n" +
                        "GROUP BY window_start,window_end,refund_time_type\n"
        );
        return tableResult;
    }

    private static void saveToSink(TableEnvironment tableEnv, Table resultTable) {
        tableEnv.createTemporaryView("resultTable", resultTable);
        tableEnv.executeSql("INSERT INTO dws_refund_after_behavior SELECT * FROM resultTable");
    }
}
